package io.kiki.sba.registry.server.metadata.lease.session;


import io.kiki.sba.registry.common.model.metaserver.Lease;
import io.kiki.sba.registry.common.model.metaserver.cluster.VersionedList;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.ClientInterfaceNode;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.lifecycle.impl.LifecycleHelper;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.cluster.node.NodeAdded;
import io.kiki.sba.registry.server.metadata.cluster.node.NodeModified;
import io.kiki.sba.registry.server.metadata.cluster.node.NodeRemoved;
import io.kiki.sba.registry.server.metadata.lease.impl.AbstractEvictableFilterableLeaseManager;
import io.kiki.sba.registry.server.metadata.monitor.Metrics;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;

@Component
public class ClientInterfaceServerManagerImpl extends AbstractEvictableFilterableLeaseManager<ClientInterfaceNode> implements ClientInterfaceNodeManager {
    private static final Logger logger = LoggerFactory.getLogger(ClientInterfaceServerManagerImpl.class);

    @Autowired
    private ServerConfig serverConfig;

    @Autowired
    private SlotManager slotManager;

    public ClientInterfaceServerManagerImpl() {
    }

    public ClientInterfaceServerManagerImpl(ServerConfig serverConfig, SlotManager slotManager, MetadataLeaderService metadataLeaderService) {
        this.serverConfig = serverConfig;
        this.slotManager = slotManager;
        this.metadataLeaderService = metadataLeaderService;
    }

    @PostConstruct
    public void postConstruct() throws Exception {
        LifecycleHelper.initializeIfPossible(this);
        LifecycleHelper.startIfPossible(this);
    }

    @PreDestroy
    public void preDestory() throws Exception {
        LifecycleHelper.stopIfPossible(this);
        LifecycleHelper.disposeIfPossible(this);
    }

    @Override
    public void register(Lease<ClientInterfaceNode> lease) {
        super.register(lease);
        notifyObservers(new NodeAdded<>(lease.getRenewal()));
    }

    /**
     * Different from data server, session node maintains a 'ProcessId' to be as unique Id for Session
     * Process(not server)
     *
     * <p>Once a restart event happened on the same session-server, an notification will be sent
     */
    @Override
    public boolean renew(ClientInterfaceNode clientInterfaceNode, int duration) {
        Metrics.Heartbeat.onSessionHeartbeat(clientInterfaceNode.getIp());
        Lease<ClientInterfaceNode> lease = getLease(clientInterfaceNode);
        if (clientInterfaceNode.getClientInterfaceId() != null//
                && lease != null//
                && lease.getRenewal() != null//
                && !Objects.equals(lease.getRenewal().getClientInterfaceId(), clientInterfaceNode.getClientInterfaceId())) {
            logger.warn("[renew] session node is restart, as process-Id change from {} to {}", lease.getRenewal().getClientInterfaceId(), clientInterfaceNode.getClientInterfaceId());
            // replace the session node, as it has changed process-id already
            lease.setRenewal(clientInterfaceNode);
            super.register(new Lease<>(clientInterfaceNode, duration));
            notifyObservers(new NodeModified<>(lease.getRenewal(), clientInterfaceNode));
            return false;
        } else {
            // replace the session node, as it has changed weight already
            if (clientInterfaceNode.getClientInterfaceId() != null && lease != null && lease.getRenewal() != null && clientInterfaceNode.getWeight() != lease.getRenewal().getWeight()) {
                lease.setRenewal(clientInterfaceNode);
            }
            return super.renew(clientInterfaceNode, duration);
        }
    }

    @Override
    public boolean cancel(Lease<ClientInterfaceNode> lease) {
        boolean result = super.cancel(lease);
        if (result) {
            Metrics.Heartbeat.onSessionEvict(lease.getRenewal().getIp());
            notifyObservers(new NodeRemoved<>(lease.getRenewal()));
        }
        return result;
    }

    @Override
    protected int getIntervalMilli() {
        return serverConfig.getExpireCheckIntervalMillis();
    }

    @Override
    protected int getEvictBetweenMilli() {
        return serverConfig.getExpireCheckIntervalMillis();
    }

    @Override
    public VersionedList<ClientInterfaceNode> getSessionServerMetaInfo() {
        VersionedList<Lease<ClientInterfaceNode>> leaseMeta = VersionedLeaseList();
        List<ClientInterfaceNode> list = new ArrayList<>();
        leaseMeta.getList().forEach(lease -> {
            list.add(lease.getRenewal());
        });
        return new VersionedList<>(list, leaseMeta.getEpoch());
    }

    @Override
    public long getEpoch() {
        return currentEpoch.get();
    }

    @Override
    public void onHeartbeat(HeartbeatRequest<ClientInterfaceNode> heartbeatRequest) {
        if (amILeader() && !metadataLeaderService.isWarmuped()) {
            learnFromSession(heartbeatRequest);
        }
    }

    protected void learnFromSession(HeartbeatRequest<ClientInterfaceNode> heartbeat) {
        Slots slots = heartbeat.getSlots();
        slotManager.refresh(slots);
    }
}
